package com.shujia.spark.streaming

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, Statement}
import scala.collection.mutable

object Demo05Bukong {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]")
      .getOrCreate()


    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))
    val textDS: DStream[String] = ssc.socketTextStream("master", 8888)

    // 使用本地集合保存黑名单，虽然使用非常简单，但是不利于更新黑名单，每次如果需要更新则程序需要重启
    //    val mdnBlacklist: Set[String] = Set("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7", "82044609CB65BA7139205E7808144D5C254C419F", "27A43BC5BB6853DE779A9EA7F3523733CACBD21C")
    // 为了方便黑名单的更新，选择从外部数据源获取数据
    // 如果直接放在算子外部，则只会在程序启动时在Driver端启动一次
    //    // 建立JDBC连接
    //    val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
    //    // 创建prepareStatement
    //    val st: Statement = conn.createStatement()
    //
    //    val rs: ResultSet = st.executeQuery("select mdn from mdn_blacklist")
    //    val mdnBlacklist: mutable.Set[String] = mutable.Set[String]()
    //
    //    while (rs.next()) {
    //      mdnBlacklist.add(rs.getString("mdn"))
    //    }

    // 基于电信基站的用户位置数据进行过滤
    textDS
      .filter(line => {
        var flag: Boolean = false

        // 如果直接将MySQL数据的获取放入算子内部，则来一条数据会建立一次连接，性能太低了
        val mdn: String = line.split(",")(0)
        // 建立JDBC连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
        // 创建prepareStatement
        val pSt: PreparedStatement = conn.prepareStatement("select mdn from mdn_blacklist where mdn = ?")
        pSt.setString(1, mdn)
        val rs: ResultSet = pSt.executeQuery()
        while (rs.next()) {
          flag = true
        }
        flag
      })
      .map(line => {
        val splits: Array[String] = line.split(",")
        // mdn city_id county_id start_time end_time
        (splits(0), splits(2), splits(3), splits(5), splits(6))
      })
    //      .print()

    /**
     * 综上：连接既不能放在Driver端，也不能直接放在算子内部
     * 所以取折中，让每个批次建立一次连接
     */

    textDS
      .transform(rdd => {
        // 通过transform将每个批次变成一个一个的RDD进行处理
        // 每个批次会建立一次连接
        // 建立JDBC连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
        // 创建prepareStatement
        val st: Statement = conn.createStatement()

        val rs: ResultSet = st.executeQuery("select mdn from mdn_blacklist")
        val mdnBlacklist: mutable.Set[String] = mutable.Set[String]()

        while (rs.next()) {
          mdnBlacklist.add(rs.getString("mdn"))
        }
        // 广播变量
        val mdnBlacklistBro: Broadcast[mutable.Set[String]] = ssc.sparkContext.broadcast(mdnBlacklist)

        rdd
          .filter(line => {
            val mdn: String = line.split(",")(0)
            mdnBlacklistBro.value.contains(mdn)
          })
      })
      .map(line => {
        val splits: Array[String] = line.split(",")
        // mdn city_id county_id start_time end_time
        (splits(0), splits(2), splits(3), splits(5), splits(6))
      })
      .print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}
